package com.doit.day06

import scala.collection.immutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.io.Source

//1.求每部电影的平均分  ==>  按照降序输出
//movie.txt   电影id 电影名称  电影类型
//rating.txt  1_1193_5_978300760    用户id，电影id，评分分数   评分的时间戳
//users.txt   1_F_1_10_48067        用户id，gender，age，职业，邮编
object _04_每部电影的平均分 {
  def main(args: Array[String]): Unit = {
    //不管啥，先把三个表里面的数据都取出来==》 进行封装
    val movies: List[String] = Source.fromFile("C:\\Users\\yi\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val ratings: List[String] = Source.fromFile("C:\\Users\\yi\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val users: List[String] = Source.fromFile("C:\\Users\\yi\\Desktop\\课程资料\\资料\\users.txt").getLines().toList

    val userAndJob: Map[String, String] = users.map(line => {
      val arr: Array[String] = line.split("_")
      val user_id: String = arr(0)
      val job: String = arr(3)
      (user_id, job)
    }).toMap


  val ratingsTuples: List[(String, String, String, String)] = ratings.map(line => {
      val arr: Array[String] = line.split("_")
      (arr(0), arr(1), arr(2), arr(3))
    })

    val tuples: Map[String, (String, String)] = movies.map(line => {
      val arr: Array[String] = line.split("_")
      (arr(0),(arr(1), arr(2)))
    }).toMap

    val buffer: ArrayBuffer[((String, String), String, String)] = new ArrayBuffer[((String, String), String, String)]()

    //8.补全信息(返回电影的名称，类型，被评的所有分集合，所有评分的人集合)
    for (elem <- ratingsTuples) {
      val movie_id: String = elem._2
      val tuple: (String, String) = tuples.getOrElse(movie_id, ("电影名称", "电影类型"))
      val user_id: String = elem._1
      val rating: String = elem._3
      buffer.append((tuple,user_id,rating))
    }

    val tupleToTuples: Map[(String, String), ArrayBuffer[((String, String), String, String)]] = buffer.groupBy(_._1)

    tupleToTuples.map(tp=>{
      val idType: (String, String) = tp._1
      val value: ArrayBuffer[((String, String), String, String)] = tp._2
      val user_ids: ArrayBuffer[String] = value.map(_._2)
      val ratings: ArrayBuffer[String] = value.map(_._3)
      (idType._1,idType._2,user_ids,ratings)
    }).foreach(println)

    //7.每个职业最喜欢的前三个电影类型(按照观看次数排)  -->多了一步   获取到用户id，然后再去关联职业==> user_id  map(user_id，职业作为value)
    //第6个需求的中间结果 ==》 电影类型，电影的评分

  /*  val buffer: ArrayBuffer[(String, String)] = new ArrayBuffer[(String, String)]()

    for (elem <- ratingsTuples) {
      val user_id: String = elem._1
      val movie_id: String = elem._2
      //根据电影id，从电影表里面获取对应的电影类型
      val tuple: (String, String) = tuples.getOrElse(movie_id, ("未知", "未知"))
      val types: String = tuple._2
      //因为电影表里面的类型是一个String，需要对他们进行切割，获取到每一个类型
      val movieTypes: Array[String] = types.split("\\|")
      val job: String = userAndJob.getOrElse(user_id, "国家电网")
      for (typ <- movieTypes) {
        buffer.append((job,typ))
      }
    }

      //截至到现在  我得到的这个buffer里面==》   每个职业最喜欢的前三个电影类型(按照观看次数排)
      //10,Comedy  我是对job进行分组  还是对类型进行分组 还是对两个合起来的元组进行分组
      //如果我对job进行分组==》 job，List((job,类型),(job，类型).....)
      val jobGrouped: Map[String, ArrayBuffer[(String, String)]] = buffer.groupBy(_._1)

      //下一步要干啥事？
      val stringToTuples: Map[String, List[(String, Int)]] = jobGrouped.map(tp => {
        val job: String = tp._1
        val value: ArrayBuffer[(String, String)] = tp._2
        val types: ArrayBuffer[String] = value.map(tp1 => {
          val movieType: String = tp1._2
          movieType
        })
        (job, types.groupBy(word => word).map(tp => (tp._1, tp._2.length)).toList.sortBy(-_._2).take(3))
      })

    stringToTuples.foreach(println)*/

    //6.最受欢迎(被评分次数最多的)的前三个电影类型
 /*   val idAndType: Map[String, Array[String]] = tuples.map(tp => {
      val movieType: Array[String] = tp._2._2.split("\\|")
      (tp._1, movieType)
    })

    //对评分表获取出来我们所关心的数据(电影id，评分)
    val idAndRating: List[(String, String)] = ratingsTuples.map(tp => (tp._2, tp._3))

    val buffer: ArrayBuffer[(String, Double)] = new ArrayBuffer[(String, Double)]()

    //for循环走完以后，就有了所有类型的所有评分
    for (tp1 <- idAndRating) {
      val movieType: Array[String] = idAndType.getOrElse(tp1._1,Array[String]())
      for (typ <- movieType) {
        buffer.append((typ,tp1._2.toDouble))
      }
    }

    //按照电影类型去分组
    buffer.groupBy(_._1)
      .map(tp=>{
        val movieTpye: String = tp._1
        val cnt: Int = tp._2.size
        (movieTpye,cnt)
      }).toList
      .sortBy(-_._2)
      .take(20)
      .foreach(println)*/
    // 5.每个电影类型的平均分按照降序排列取前3
/*    val idAndType: Map[String, Array[String]] = tuples.map(tp => {
      val movieType: Array[String] = tp._2._2.split("\\|")
      (tp._1, movieType)
    })

    //对评分表获取出来我们所关心的数据(电影id，评分)
    val idAndRating: List[(String, String)] = ratingsTuples.map(tp => (tp._2, tp._3))

    val buffer: ArrayBuffer[(String, Double)] = new ArrayBuffer[(String, Double)]()

    //for循环走完以后，就有了所有类型的所有评分
    for (tp1 <- idAndRating) {
          val movieType: Array[String] = idAndType.getOrElse(tp1._1,Array[String]())
          for (typ <- movieType) {
            buffer.append((typ,tp1._2.toDouble))
          }
    }

    val typeAndRat: Map[String, ArrayBuffer[(String, Double)]] = buffer.groupBy(_._1)
    //求平均分
    val typeAndT: Map[String, Double] = typeAndRat.map(tp => {
      (tp._1, (tp._2.map(_._2).sum / tp._2.size).formatted("%.2f").toDouble)
    })

    typeAndT.toList.sortBy(-_._2).take(10).foreach(println)*/

    /*
      //4.哪些年份的电影评分(平均分)最高，取最高的前五个年份
      // ==》   年份   平均分弄出来   排序，取前5
      // ==》 用substring 将电影名称中的年份单独截取出来

      val movie_idAndYear: Map[String, String] = tuples.map(tp => {
        val movie_id: String = tp._1
        val movie_name: String = tp._2._1
        val year: String = movie_name.substring(movie_name.length - 5, movie_name.length - 1)
        (movie_id, year)
      }).toMap

      //现在有了电影的id，但是没有电影id的评分 ==》 按照电影id分类，求出他的总平均分==》
      // 1 2 3 4 5 ==》 3    10 11 ==》 5.5    8.5/2 4.25     36/7   这样是不对的
      //把所有的评分都放在一个集合中   求了总分   求总个数
      //(电影id，评分)  ==》 map   get() 补全吗？   评分表做成这个 电影id  分数的集合
      //==》 groupBy   collect_list
      val id_tuple: Map[String, List[(String, String, String, String)]] = ratingsTuples.groupBy(_._2)
      val idAndRating: immutable.Seq[(String, List[String])] = id_tuple.map(tp => {
        val value: List[(String, String, String, String)] = tp._2
        val ratings: List[String] = value.map(_._3)
        (tp._1, ratings)
      }).toList

     val res: immutable.Seq[(String, List[String])] =  for (elem <- idAndRating) yield{
        val movie_id: String = elem._1
        val year: String = movie_idAndYear.getOrElse(movie_id, "9999")
        (year,elem._2)
      }

      //求平均分
      val stringToTuples: Map[String, immutable.Seq[(String, List[String])]] = res.groupBy(_._1)

      val list: List[(String, Double)] = stringToTuples.map(tp => {
        val year: String = tp._1
        val value: immutable.Seq[(String, List[String])] = tp._2
        val seq: immutable.Seq[List[Double]] = value.map(tp => {
          tp._2.map(_.toDouble)
        })
        val flatten: immutable.Seq[Double] = seq.flatten
        (year, (flatten.sum / flatten.length).formatted("%.2f").toDouble)
      }).toList
      list.sortBy(-_._2).take(5).foreach(println)*/
    //3.每个用户评价的前三部电影(用户，电影信息，评分值)==>user_id,movie_name,movie_type,ratings==> 看最早的时间
    //先从rating表中获取到每个用户评论过的所有的电影，进行排序.take(3)
    //按照用户id进行分组，得到一个map类型，key就是用户id，value是该用户评分过的所有的电影集合
 /*   val userAndTop3Movie: Map[String, List[(String, String, String, String)]] = ratingsTuples.groupBy(_._1)
      .map(tp => {
        val user_id: String = tp._1
        val top3Movie: List[(String, String, String, String)] = tp._2.sortBy(_._4.toLong).take(3)
        (user_id, top3Movie)
      })

    val res: ArrayBuffer[(String, String, String, String, String,String)] = new ArrayBuffer[(String, String, String, String, String,String)]

    for (elem <- userAndTop3Movie) {
      val top3Movie: List[(String, String, String, String)] = elem._2
      for (tp <- top3Movie) {
        //(用户id，电影id，评分，时间戳)
        val movienameAndType: (String, String) = tuples.getOrElse(tp._2, ("没找到", "没找到"))
        res.append((tp._1,tp._2,movienameAndType._1,movienameAndType._2,tp._3,tp._4))
      }
    }
    res.filter(_._1.equals("5100")).foreach(println)*/
    //获取到我们想要的数据  电影的平均分==》 这部电影的总分和评论的总次数
    //我需要将电影id转换成电影名称
   /* val movie_id_avg_ratings: Map[String, Double] = ratingsTuples.groupBy(_._2)
      .map(tp => {
        val movie_id: String = tp._1
        val movie_name: String = tuples.getOrElse(movie_id, ("没找到", "没找到"))._1
        (movie_name, (tp._2.map(_._3.toDouble).sum / tp._2.size).formatted("%.2f").toDouble)
      })

    movie_id_avg_ratings
      .toList
      .sortBy(-_._2)
      .take(20)
      .foreach(println)*/
/*2.求每个用户评过电影的总数，按照降序排列
    ratingsTuples.groupBy(_._1)
      .map(tp=>(tp._1,tp._2.size))
      .toList
      .sortBy(-_._2)
      .take(20)
      .foreach(println)*/
  }
}
